package com.chen.pipeline;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.chen.util.ChineseToSpell;
import com.chen.util.DbUtil;
import org.apache.commons.lang3.StringUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.pipeline.Pipeline;

import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.*;

public class JDItemPipeline implements Pipeline {
    private static JDItemPipeline jdItemPipeline;
    private Set<String> AllType = new HashSet<>();

    public static JDItemPipeline getInstance() {
        return jdItemPipeline;
    }

    public static JDItemPipeline getInstance(JedisPool pool) {
        if (jdItemPipeline == null) {
            jdItemPipeline = new JDItemPipeline(pool);
        }
        return jdItemPipeline;
    }


    private ExecutorService executorService = Executors.newSingleThreadExecutor(r -> {
        Thread thread = new Thread(r);
        thread.setName("[sendThread]");
        return thread;
    });

    public boolean shutDown() {
        pool.close();
        executorService.shutdown();
        return executorService.isShutdown() & pool.isClosed();
    }

    private final BlockingQueue<Map<String, Object>> datas = new ArrayBlockingQueue<>(1000);

    protected JedisPool pool;

    private final Queue<Map<String, Object>> waitTosend = new LinkedBlockingDeque<>();

    private JDItemPipeline(JedisPool pool) {
        this.pool = pool;
        jdItemPipeline = this;
    }

    private String getDATALIST(Task task, Map<String, Object> map) {
        return task.getUUID() + "_Type_" + ChineseToSpell.getFullSpell((String) map.get("type"));
    }

    private void copyDatas() {
        if (waitTosend.addAll(datas))
            datas.clear();
    }

    private void sendDatas(Task task) {
        Jedis resource = pool.getResource();
        try {
            Map<String, String> tableInfo = new HashMap<>();
            long a = System.currentTimeMillis();
            redis.clients.jedis.Pipeline pipelined = resource.pipelined();
            pipelined.select(1);
            Map<String, Object> temp;
            while ((temp = waitTosend.poll()) != null) {
                tableInfo.put(getDATALIST(task, temp), getTypeInfo(temp));
                pipelined.rpush(getDATALIST(task, temp), JSON.toJSONString(temp));
            }
            Iterator<Map.Entry<String, String>> iterator = tableInfo.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, String> next = iterator.next();
                if (AllType.add(next.getKey())) {
                    pipelined.hset("Table_info", next.getKey(), next.getValue());
                }
            }
            pipelined.sync();
            System.out.println(Thread.currentThread().getName() + " 数据插入完毕,耗时：" + (System.currentTimeMillis() - a) + "剩余数据：" + waitTosend.size());
            DbUtil.typeInfoAdd(tableInfo.entrySet());
        } catch (SQLException throwables) {
            System.out.println("已有");
        } finally {
            resource.close();
        }
    }

    private String getTypeInfo(Map temp) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("type", temp.get("type").toString());
        jsonObject.put("state", 0);
        return jsonObject.toJSONString();
    }

    static public void updateType() {
        getInstance().executorService.execute(() -> {
            List<String> except = null;
            try {
                except = DbUtil.getExcept();
                for (String s : except) {
                    System.out.println("更新" + s + "状态为1");
                    getInstance().setTypeInfo(s);
                }
            } catch (SQLException throwables) {
                throwables.printStackTrace();
            }
        });
    }

    public void setTypeInfo(String type) {
        Jedis resource = pool.getResource();
        try {
            resource.select(1);
            String table_info = resource.hget("Table_info", type);
            if (StringUtils.isBlank(table_info)) {
                System.out.println("没有查找到种类");
                return;
            }
            JSONObject jsonObject = JSON.parseObject(table_info);
            jsonObject.put("state", 1);
            resource.hset("Table_info", type, jsonObject.toJSONString());
        } finally {
            resource.close();
        }
    }


    @Override
    public void process(ResultItems resultItems, Task task) {
        if (resultItems.get("type") != null && resultItems.get("type").equals("item")) {
            List<Map<String, Object>> data = resultItems.get("data");
            try {
                for (Map<String, Object> datum : data) {
                    if (datas.size() == 1000) {
                        copyDatas();
                        executorService.execute(() -> {
                            System.out.println(Thread.currentThread().getName() + " 开始请求发送信息，剩余：" + waitTosend.size());
                            sendDatas(task);
                        });
                    }
                    datas.offer(datum, 100, TimeUnit.SECONDS);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
